"""
Case Type   : Ustore_Standby_Logicreplication
Case Name   : 复制槽操作权限验证，新增用户没有replication权限，不在白名单里
Create At   : 2022/03/17
Owner       : zhao-pengcheng-g
Description :
    1.备份postgresql.conf,pg_hba.conf文件
    2.修改ustore参数
    3.修改逻辑复制相关参数
    4.创建用户，新增用户未设置权限和白名单,配置默认用户白名单
    5.主节点默认用户创建逻辑复制槽1
    6.新增用户创建复制槽2
    7.主备节点新增用户启动复制槽1
    8.新增用户删除复制槽1
    9.删除逻辑复制槽
    10.资源清理，恢复参数
Expect      :
    1.备份成功
    2.修改ustore参数成功
    3.修改逻辑复制相关参数成功
    4.创建成功，修改pg_hba文件成功
    5.创建逻辑复制槽成功
    6.新增用户创建失败
    7.新增用户启动复制槽1失败
    8.新增用户删除复制槽1失败
    9.删除逻辑复制槽成功
    10.资源清理成功，恢复参数成功
History     :
"""
import os
import time
import unittest
from yat.test import Node
from yat.test import macro
from testcase.utils.Logger import Logger
from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.ComThread import ComThread

Primary_SH = CommonSH('PrimaryDbUser')

@unittest.skipIf(2 != Primary_SH.get_node_num(),'非1+1环境不执行')
class StandbyLogicReplication(unittest.TestCase):
    def setUp(self):
        self.logger = Logger()
        self.logger.info(f"-----{os.path.basename(__file__)} start-----")
        self.primary_node = Node('PrimaryDbUser')
        self.standby_node = Node('Standby1DbUser')
        self.sh_primy = CommonSH('PrimaryDbUser')
        self.sh_standby = CommonSH('Standby1DbUser')
        self.constant = Constant()
        self.common = Common()
        self.dir_replication = os.path.join(macro.DB_BACKUP_PATH,
                                            'dir_replication_0052')
        self.postgresql_path = os.path.join(macro.DB_INSTANCE_PATH,
                                            macro.DB_PG_CONFIG_NAME)
        self.postgresql_bak_path = os.path.join(self.dir_replication,
                                                'postgresql_bak.conf')
        self.pg_hba_path = os.path.join(macro.DB_INSTANCE_PATH,
                                        macro.PG_HBA_FILE_NAME)
        self.pg_hba_bak_path = os.path.join(self.dir_replication,
                                            'pg_hba_bak.conf')
        self.db_name = 'db_replication_0052'
        self.user_name = 'u_logic_replication_0052'
        self.slot_name_1 = 'slot_standby_logic_replication_0052_1'
        self.slot_name_2 = 'slot_standby_logic_replication_0052_2'
        self.log_file_1 = os.path.join(self.dir_replication,
                                       'standby_logic_replication_0052_1.log')

        text = "-----step1:备份postgresql.conf,pg_hba.conf;expect:成功-----"
        self.logger.info(text)
        cp_cmd = f'''rm -rf {self.dir_replication} && 
                    mkdir {self.dir_replication} &&
                    cp {self.postgresql_path} {self.postgresql_bak_path} && 
                    cp {self.pg_hba_path} {self.pg_hba_bak_path} && echo OK'''
        cp_pri_msg = self.primary_node.sh(cp_cmd).result()
        self.logger.info(cp_pri_msg)
        self.assertEqual("OK", cp_pri_msg, "执行失败" + text)
        cp_sta_msg = self.standby_node.sh(cp_cmd).result()
        self.logger.info(cp_sta_msg)
        self.assertEqual("OK", cp_sta_msg, "执行失败" + text)

        text = "-----step2:修改ustore参数;expect:成功-----"
        self.logger.info(text)
        self.config_ustore = 'enable_default_ustore_table'
        self.show_ustore_cmd = f'show {self.config_ustore};'
        check_res = self.sh_primy.execut_db_sql(self.show_ustore_cmd)
        self.logger.info(check_res)
        self.ustore_default = check_res.splitlines()[-2].strip()
        if 'on' != self.ustore_default:
            guc_result = self.sh_primy.execute_gsguc('reload',
                                            self.constant.GSGUC_SUCCESS_MSG,
                                            f'{self.config_ustore}=on')
            self.assertTrue(guc_result, '执行失败' + text)
        show_msg = self.sh_primy.execut_db_sql(self.show_ustore_cmd)
        self.logger.info(show_msg)
        self.common.equal_sql_mdg(show_msg, self.config_ustore, 'on',
                                  '(1 row)', flag='1')

        text = "-----step3:修改逻辑复制相关参数;expect:成功-----"
        self.logger.info(text)
        self.config_slot_log = 'enable_slot_log'
        self.show_slot_log_cmd = f'show {self.config_slot_log};'
        check_res = self.sh_primy.execut_db_sql(self.show_slot_log_cmd)
        self.logger.info(check_res)
        self.slot_log_default = check_res.splitlines()[-2].strip()
        if 'on' != self.slot_log_default:
            guc_result = self.sh_primy.execute_gsguc('reload',
                                            self.constant.GSGUC_SUCCESS_MSG,
                                            f'{self.config_slot_log}=on')
            self.assertTrue(guc_result, '执行失败' + text)
        show_msg = self.sh_primy.execut_db_sql(self.show_slot_log_cmd)
        self.logger.info(show_msg)
        self.common.equal_sql_mdg(show_msg, self.config_slot_log,
                                  'on', '(1 row)', flag='1')

        self.config_wal_level = 'wal_level'
        self.show_wal_level_cmd = f'show {self.config_wal_level};'
        check_res = self.sh_primy.execut_db_sql(self.show_wal_level_cmd)
        self.logger.info(check_res)
        self.wal_level_default = check_res.splitlines()[-2].strip()
        if 'logical' != self.wal_level_default:
            guc_result = self.sh_primy.execute_gsguc('set',
                                            self.constant.GSGUC_SUCCESS_MSG,
                                            f'{self.config_wal_level}=logical')
            self.assertTrue(guc_result, '执行失败' + text)
            restart_msg = self.sh_primy.restart_db_cluster()
            self.logger.info(restart_msg)
            self.assertTrue(restart_msg, '重启失败')
        show_msg = self.sh_primy.execut_db_sql(self.show_wal_level_cmd)
        self.logger.info(show_msg)
        self.common.equal_sql_mdg(show_msg, self.config_wal_level,
                                  'logical', '(1 row)', flag='1')
        status = self.sh_primy.get_db_cluster_status()
        self.assertNotIn("ERROR", status, '执行失败:' + text)
        self.assertIn("Normal", status, '执行失败:' + text)

        text = "-----step4:创建用户，新增用户未设置权限和白名单," \
               "配置默认用户白名单;expect:成功-----"
        self.logger.info(text)
        create_cmd = f'''drop user if exists {self.user_name};
                    create user {self.user_name} with \
                    password '{macro.COMMON_PASSWD}';'''
        create_msg = self.sh_primy.execut_db_sql(create_cmd,
                                                 dbname='postgres')
        self.logger.info(create_msg)
        self.assertIn(self.constant.CREATE_ROLE_SUCCESS_MSG,
                      create_msg, '执行失败' + text)
        param_1 = f'local replication {self.primary_node.ssh_user} trust'
        param_2 = f'host replication {self.primary_node.ssh_user} 127.0.0.1/32 trust'
        param_3 = f'host replication {self.primary_node.ssh_user} ::1/128 trust'
        result_1 = self.sh_primy.execute_gsguc('reload',
                                               self.constant.GSGUC_SUCCESS_MSG,
                                               '', pghba_param=param_1)
        result_2 = self.sh_primy.execute_gsguc('reload',
                                               self.constant.GSGUC_SUCCESS_MSG,
                                               '', pghba_param=param_2)
        result_3 = self.sh_primy.execute_gsguc('reload',
                                               self.constant.GSGUC_SUCCESS_MSG,
                                               '', pghba_param=param_3)
        self.assertTrue(result_1 and result_2 and result_3, '执行失败' + text)
        restart_msg = self.sh_primy.restart_db_cluster()
        self.logger.info(restart_msg)
        self.assertTrue(restart_msg, '重启失败')

    def test_standby_logicreplication(self):
        text = "-----step5:主节点默认用户创建逻辑复制槽1;expect:成功-----"
        self.logger.info(text)
        create_slot_1 = f"pg_recvlogical -o skip-empty-xacts=true " \
            f"-d postgres -S {self.slot_name_1} " \
            f"-p {self.primary_node.db_port} " \
            f"-U {self.primary_node.ssh_user} " \
            f"-s 2 --create"
        execute_cmd = f'''source {macro.DB_ENV_PATH};
                        expect <<EOF
                        set timeout 300
                        spawn {create_slot_1}
                        expect "Password:"
                        send "{macro.COMMON_PASSWD}\\n"
                        expect eof\n''' + '''EOF'''
        self.logger.info(execute_cmd)
        execute_msg = self.primary_node.sh(execute_cmd).result()
        self.logger.info(execute_msg)
        text = "-----验证创建逻辑复制槽成功;expect:成功-----"
        self.logger.info(text)
        sql_cmd = f"select * from pg_get_replication_slots();"
        sql_msg = self.sh_primy.execut_db_sql(sql_cmd, dbname='postgres')
        self.logger.info(sql_msg)
        self.assertIn(self.slot_name_1, sql_msg, '执行失败:' + text)

        text = "-----step6:新增用户创建复制槽2;expect:新增用户创建复制槽失败-----"
        self.logger.info(text)
        create_slot_2 = f"source {macro.DB_ENV_PATH};" \
            f"pg_recvlogical -o skip-empty-xacts=true " \
            f"-d postgres -S {self.slot_name_2} " \
            f"-p {self.primary_node.db_port} -s 2 " \
            f"-h {self.primary_node.db_host} -U {self.user_name} " \
            f"--create "
        self.logger.info(create_slot_2)
        create_slot_msg = self.primary_node.sh(create_slot_2).result()
        self.logger.info(create_slot_msg)
        self.assertIn("no pg_hba.conf entry for replication connection",
                      create_slot_msg, '执行失败:' + text)

        text = "-----step7:主备节点新增用户启动复制槽1;" \
               "expect:新增用户启动复制槽1失败-----"
        self.logger.info(text)
        start_primary = f"source {macro.DB_ENV_PATH};" \
            f"pg_recvlogical -o skip-empty-xacts=true -F 1 -d postgres " \
            f"-S {self.slot_name_1} -p {self.primary_node.db_port} " \
            f"-h {self.primary_node.db_host} -U {self.user_name} " \
            f"-s 2 -f {self.log_file_1} --start"
        self.logger.info(start_primary)
        start_primary_msg = self.primary_node.sh(start_primary).result()
        self.logger.info(start_primary_msg)
        self.assertIn("no pg_hba.conf entry for replication connection",
                      start_primary_msg, '执行失败:' + text)
        start_standy = f"source {macro.DB_ENV_PATH};" \
            f"pg_recvlogical -o skip-empty-xacts=true -F 1 -d postgres " \
            f"-S {self.slot_name_1} -p {self.primary_node.db_port} " \
            f"-h {self.standby_node.db_host} -U {self.user_name} " \
            f"-s 2 -f {self.log_file_1} --start"
        self.logger.info(start_standy)
        start_standy_msg = self.primary_node.sh(start_standy).result()
        self.logger.info(start_standy_msg)
        self.assertIn("no pg_hba.conf entry for replication connection",
                      start_standy_msg, '执行失败:' + text)

        text = "-----step8:新增用户删除复制槽1;" \
               "expect:新增用户删除复制槽1失败-----"
        self.logger.info(text)
        drop_primary = f"source {macro.DB_ENV_PATH};" \
            f"pg_recvlogical -o skip-empty-xacts=true -d postgres " \
            f"-S {self.slot_name_1} -p {self.primary_node.db_port} " \
            f"-s 2 -h {self.primary_node.db_host} --drop -U {self.user_name}"
        self.logger.info(drop_primary)
        drop_primary_msg = self.primary_node.sh(drop_primary).result()
        self.logger.info(drop_primary_msg)
        self.assertIn("no pg_hba.conf entry for replication connection",
                      drop_primary_msg, '执行失败:' + text)
        
    def tearDown(self):
        text = "-----step9:删除逻辑复制槽;expect:成功-----"
        self.logger.info(text)
        drop_slot_cmd_1 = f"pg_recvlogical -d postgres " \
            f"-S {self.slot_name_1} " \
            f"-p {self.primary_node.db_port} " \
            f"-U {self.primary_node.ssh_user} " \
            f"-s 2 --drop"
        execute_cmd = f'''source {macro.DB_ENV_PATH};
                        expect <<EOF 
                        set timeout 300 
                        spawn {drop_slot_cmd_1} 
                        expect "Password:" 
                        send "{macro.COMMON_PASSWD}\\n" 
                        expect eof\n''' + '''EOF'''
        self.logger.info(execute_cmd)
        execute_msg = self.primary_node.sh(execute_cmd).result()
        self.logger.info(execute_msg)
        self.logger.info("-----验证是否删除成功;expect:成功-----")
        sql_cmd = f"select slot_name,plugin,slot_type,active," \
            f"restart_lsn from pg_get_replication_slots();"
        sql_msg = self.sh_primy.execut_db_sql(sql_cmd, dbname='postgres')
        self.logger.info(sql_msg)
        
        text = "-----step10:资源清理，恢复参数，expect:成功-----"
        self.logger.info(text)
        text_1 = "-----step10.1:删除用户，expect:成功-----"
        self.logger.info(text_1)
        drop_cmd = f'''drop user {self.user_name};'''
        drop_msg = self.sh_primy.execut_db_sql(drop_cmd, dbname='postgres')
        self.logger.info(drop_msg)

        text_2 = "-----step10.2:恢复postgresql.conf,pg_hba.conf文件，" \
                 "expect:成功-----"
        self.logger.info(text_2)
        restore_cmd_1 = f'''
            if [[ -e {self.postgresql_path} && -e {self.postgresql_bak_path} ]]; 
            then rm -rf {self.postgresql_path}; 
            mv {self.postgresql_bak_path} {self.postgresql_path}; 
            echo OK;
            else echo not exits; fi'''
        self.logger.info(restore_cmd_1)
        restore_primary_1 = self.primary_node.sh(restore_cmd_1).result()
        restore_standby_1 = self.standby_node.sh(restore_cmd_1).result()
        self.logger.info(restore_primary_1)
        self.logger.info(restore_standby_1)
        restore_cmd_2 = f'''
            if [[ -e {self.pg_hba_path} && -e {self.pg_hba_bak_path} ]];
            then rm -rf {self.pg_hba_path};  
            mv {self.pg_hba_bak_path} {self.pg_hba_path};
            echo OK;
            else echo not exits; fi'''
        self.logger.info(restore_cmd_2)
        restore_primary_2 = self.primary_node.sh(restore_cmd_2).result()
        restore_standby_2 = self.standby_node.sh(restore_cmd_2).result()
        self.logger.info(restore_primary_2)
        self.logger.info(restore_standby_2)

        text_3 = "-----step10.3:删除文件，expect:成功-----"
        self.logger.info(text_3)
        del_cmd = f'rm -rf {self.dir_replication};'
        self.logger.info(del_cmd)
        self.primary_node.sh(del_cmd)
        self.standby_node.sh(del_cmd)
        file_cmd = f'''if [ -d {self.dir_replication} ]; 
            then echo "exists"; else echo "not exists"; fi'''
        primy_file_msg = self.primary_node.sh(file_cmd).result()
        standy_file_msg = self.standby_node.sh(file_cmd).result()
        self.logger.info(primy_file_msg)
        self.logger.info(standy_file_msg)

        self.assertNotIn(self.slot_name_1, sql_msg, '执行失败:' + text)
        self.assertIn(self.constant.DROP_ROLE_SUCCESS_MSG,
                      drop_msg, '执行失败' + text_1)
        self.assertEqual('OK', restore_primary_1, '执行失败' + text_2)
        self.assertEqual('OK', restore_standby_1, '执行失败' + text_2)
        self.assertEqual('OK', restore_primary_2, '执行失败' + text_2)
        self.assertEqual('OK', restore_standby_2, '执行失败' + text_2)
        self.assertEqual("not exists", primy_file_msg, '执行失败' + text_3)
        self.assertEqual("not exists", standy_file_msg, '执行失败' + text_3)
        self.logger.info(f"-----{os.path.basename(__file__)} end-----")